package tman.system.peer.tman;

import common.configuration.TManConfiguration;
import common.peer.PeerAddress;
import java.util.ArrayList;

import cyclon.system.peer.cyclon.CyclonPartnersPort;
import cyclon.system.peer.cyclon.CyclonPartnersRequest;
import cyclon.system.peer.cyclon.CyclonPartnersResponse;

import java.util.HashSet;
import java.util.Random;
import java.util.Set;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.network.Network;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.Timer;

import tman.simulator.snapshot.Snapshot;

public final class TMan extends ComponentDefinition {

    Negative<TManPartnersPort> tmanPartnersPort = negative(TManPartnersPort.class);
    Positive<CyclonPartnersPort> cyclonPartnersPort = positive(CyclonPartnersPort.class);
    Positive<Network> networkPort = positive(Network.class);
    Positive<Timer> timerPort = positive(Timer.class);
    private long period;
    private PeerAddress self;
    private PeerAddress succ;
    private PeerAddress pred;
    private ArrayList<PeerAddress> tmanPartners;
    //private SortedMap<Integer, PeerAddress> tmanPartners;
    private ArrayList<PeerAddress> cyclonPartners;
    private TManConfiguration tmanConfiguration;
    private int c;
    private int k;

//-------------------------------------------------------------------	
    public TMan() {
        //tmanPartners = new ArrayList<PeerAddress>();
        cyclonPartners = new ArrayList<PeerAddress>();
        tmanPartners = new ArrayList<PeerAddress>() {
        };
        subscribe(handleInit, control);
        subscribe(handleRound, timerPort);
        subscribe(handleCyclonPartnersResponse, cyclonPartnersPort);
        subscribe(handleTManViewRequest, networkPort);
        subscribe(handleTManViewResponse, networkPort);
        subscribe(handleTManPartnersRequest, tmanPartnersPort);
    }
//-------------------------------------------------------------------	
    Handler<TManInit> handleInit = new Handler<TManInit>() {

        @Override
        public void handle(TManInit init) {
            self = init.getSelf();
            tmanConfiguration = init.getConfiguration();
            tmanPartners = new ArrayList<PeerAddress>();
            cyclonPartners = new ArrayList<PeerAddress>();
            period = tmanConfiguration.getPeriod();
            succ = self;
            pred = self;
            c = init.getC();
            Snapshot.setC(c);
            k = 1;



            SchedulePeriodicTimeout rst = new SchedulePeriodicTimeout(period, period);
            rst.setTimeoutEvent(new TManSchedule(rst));
            trigger(rst, timerPort);

        }
    };
    //-------------------------------------------------------------------	
    Handler<CyclonPartnersResponse> handleCyclonPartnersResponse = new Handler<CyclonPartnersResponse>() {

        @Override
        public void handle(CyclonPartnersResponse event) {
            cyclonPartners = event.getPartners();
            //System.out.println("tman: " + self + " --> " + cyclonPartners);
            Snapshot.updateCyclonPartners(self, cyclonPartners);


        }
    };
//-------------------------------------------------------------------	
    Handler<TManSchedule> handleRound = new Handler<TManSchedule>() {

        @Override
        public void handle(TManSchedule event) {
            CyclonPartnersRequest cyclonRequest = new CyclonPartnersRequest();
            trigger(cyclonRequest, cyclonPartnersPort);

            if (tmanPartners.isEmpty()) {
                if (!cyclonPartners.isEmpty()) {
                    Set cyclonSet = new HashSet<PeerAddress>(cyclonPartners);
                    tmanPartners = selectView(cyclonSet);
                    //System.err.println("Peer: " + self + " -> tmanPartners = cyclonPartners = " + cyclonPartners);
                } else {
                    return;
                    //wait for cyclon tmanViewResponse
                    //System.err.println("Peer: " + self + " -> tmanPartners and cyclonPartners empty");
                }
            }

            PeerAddress q = selectPeer();
            Set buf = new HashSet<PeerAddress>();



            buf.add(self);
            buf.addAll(tmanPartners);
            buf.addAll(cyclonPartners);
            ArrayList<PeerAddress> list = new ArrayList<PeerAddress>(buf);
            //System.err.println("Peer: " + self + " -> list to send to " + q + " : " + list);

            TManViewRequest tmanViewRequest = new TManViewRequest(null, self, q, list);
            trigger(tmanViewRequest, networkPort);
            buf.clear();


        }
    };
//-------------------------------------------------------------------	
    Handler<TManViewRequest> handleTManViewRequest = new Handler<TManViewRequest>() {

        @Override
        public void handle(TManViewRequest event) {
            Set buf = new HashSet<PeerAddress>();

            //buf.addAll(event.getView());
            buf.addAll(tmanPartners);
            buf.add(self);
            buf.addAll(cyclonPartners);

            TManViewResponse tmanViewResponse = new TManViewResponse(null, self, event.getPeerSource(), new ArrayList<PeerAddress>(buf));
            buf.clear();
            trigger(tmanViewResponse, networkPort);


            
            buf.addAll(tmanPartners);
            buf.addAll(event.getView());
            //adding the rnd.view is not specified by the tman protocol
            //buf.addAll(cyclonPartners);

            //System.out.println("tman: " + self + " --> " + cyclonPartners);
            
            tmanPartners = selectView(buf);
            buf.clear();
            updateTopology();
            
        }
    };
    //-------------------------------------------------------------------	
    Handler<TManViewResponse> handleTManViewResponse = new Handler<TManViewResponse>() {

        @Override
        public void handle(TManViewResponse event) {
            Set buf = new HashSet<PeerAddress>();
            buf.addAll(tmanPartners);
            buf.addAll(event.getView());
            //adding the rnd.view is not specified by the tman protocol
            //buf.addAll(cyclonPartners);

            //System.err.println("Peer: " + self + " -> buf :" + buf);
            
            tmanPartners = selectView(buf);

            //System.err.println("Peer: " + self + " -> tmanPartners :" + tmanPartners);
            buf.clear();
            updateTopology();
            

        }
    };
    //-------------------------------------------------------------------	
    Handler<TManPartnersRequest> handleTManPartnersRequest = new Handler<TManPartnersRequest>() {

        @Override
        public void handle(TManPartnersRequest event) {
            ArrayList<PeerAddress> listPartners = new ArrayList<PeerAddress>(tmanPartners);
            TManPartnersResponse response = new TManPartnersResponse(listPartners, succ, pred);

            trigger(response, tmanPartnersPort);
        }
    };

    private ArrayList<PeerAddress> selectView(Set<PeerAddress> buf) {
        buf.remove(self);
        //rank based on the distance (to create a ring)
        //Map<Integer, PeerAddress> tmpView = RingBuilder.rank(self, buf);
        
        //ArrayList<PeerAddress> tmpView = RingBuilder.rank(self, buf);

        //rank based on the distance prob function (to create a SWN)
        ArrayList<PeerAddress> tmpView = SWNBuilder.rank(self, buf);

        //rewire after ranking 
        //tmpView=SWNBuilder.rewire(self,tmpView,k++);

        //Set rankedSet = new HashSet<PeerAddress>();

        //add previous partners
        //rankedSet.addAll(tmanPartners);


        //pick first c from the ranked list
        /*Iterator iterator = tmpView.keySet().iterator();
        int i = 0;
        while (iterator.hasNext() & i++ < c) {
        Integer dist = (Integer) iterator.next();
        rankedSet.add(tmpView.get(dist));
        }*/

        if (tmpView.size() > c) {
            tmpView = new ArrayList<PeerAddress>(tmpView.subList(0, c));
        }
        //System.err.println("TMan -> Peer " + self + " : ViewList = " + tmpView);

        
        return tmpView;
    }

    private PeerAddress selectPeer() {
        if (tmanPartners.isEmpty()) {
            return null;
        } else if (tmanPartners.size() == 1) {
            return tmanPartners.get(0);
        } else {

            Random random = new Random();
            int q = random.nextInt(tmanPartners.size() / 2);

            //rank tmanPartners
            //ArrayList<PeerAddress> tmpView = RingBuilder.rank(self, new HashSet<PeerAddress> (tmanPartners));
            //PeerAddress peerSelected = tmpView.get(q);
            PeerAddress peerSelected = tmanPartners.get(q);

            return peerSelected;
        }
    }

    private void updateTopology() {

        RingNode[] succPred;

        try {
            succPred = RingBuilder.succPred(self, tmanPartners);
            succ = succPred[0].getSelf();
            pred = succPred[1].getSelf();
        } catch (ExceptionViewEmpty ex) {
            //Logger.getLogger(TMan.class.getName()).log(Level.SEVERE, null, ex);
            succ = null;
            pred = null;
        }

        Snapshot.updateTManPartners(self, tmanPartners);
        Snapshot.updateSuccPred(self, succ, pred);
        /*
        if (tmanPartnersList.size() == 2) {
        succ = tmanPartnersList.get(0);
        pred = tmanPartnersList.get(1);
        } else {
        succ = null;
        pred = null;
        }*/

        

    }
}
